package org.ws.task.item.support;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.item.AbstractTask;
import org.ws.task.item.PullTask;
import org.ws.task.listener.item.support.DefaultItemCommitListener;
import org.ws.task.model.protocol.*;
import org.ws.task.repository.NodeMetadataRepository;
import org.ws.task.model.TaskItem;

import java.util.Collection;

/**
 * 拉数据
 * @param <T>
 */
public class SimplePullTask<T> extends AbstractTask implements PullTask<T> {

    public SimplePullTask(String name) {
        super(name);
        this.registerListener(new DefaultItemCommitListener());
    }

    @Override
    public TaskItem<T> poll() {
        load();
        return this.context.poll();
    }

    @Override
    public TaskItem<T> element() {
        load();
        return (TaskItem<T>) this.context.element();
    }

    @Override
    public TaskItem<T> peek() {
        load();
        return this.context.peek();
    }

    /**
     * 分级拉取
     * 为空：同步
     * 不为空：异步
     */
    private synchronized void load(){
        this.awaitStart(30_000);
        long timeout = this.taskInstance.getTaskConfig().getFetchTimeout();
        if(this.context.isEmpty()){
            PullItemResponse<T> itemResponse = this.remoteAccessApi.fetchItemSync(makeItemRequest(), timeout);
            if(itemResponse != null){
                Collection<TaskItem<T>> items = itemResponse.getItems();
                this.context.addAll(items);
            }
        }

    }

    private PullItemRequest makeItemRequest(){

        Integer chunk = this.taskInstance.getTaskConfig().getChunk();
        PullItemRequest request = new PullItemRequest();
        request.setSize(chunk);
        request.setTaskName(this.name());
        request.setNodeId(NodeMetadataRepository.getInstance().getTheWorkerNode().getId());
        request.setTimeMillis(System.currentTimeMillis());
        return request;
    }
}
